/*
 * Copyright (c) 2024, Alibaba Cloud;
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.aliyun.dataworks.migrationx.transformer.core.sqoop;

import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.OptionBuilder;
import org.apache.commons.cli.OptionGroup;
import org.apache.hadoop.util.ToolRunner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.Arrays;
import java.util.Properties;

public abstract class BaseSqoopTool extends SqoopTool {

    public static final Logger LOG = LoggerFactory.getLogger(
        BaseSqoopTool.class.getName());

    public static final String HELP_STR = "\nTry --help for usage instructions.";

    // Here are all the arguments that are used by the standard sqoop tools.
    // Their names are recorded here so that tools can share them and their
    // use consistently. The argument parser applies the leading '--' to each
    // string.
    public static final String CONNECT_STRING_ARG = "connect";
    public static final String CONN_MANAGER_CLASS_NAME =
        "connection-manager";
    public static final String CONNECT_PARAM_FILE = "connection-param-file";
    public static final String DRIVER_ARG = "driver";
    public static final String USERNAME_ARG = "username";
    public static final String PASSWORD_ARG = "password";
    public static final String PASSWORD_PROMPT_ARG = "P";
    public static final String PASSWORD_PATH_ARG = "password-file";
    public static final String PASSWORD_ALIAS_ARG = "password-alias";
    public static final String DIRECT_ARG = "direct";
    public static final String BATCH_ARG = "batch";
    public static final String TABLE_ARG = "table";
    public static final String STAGING_TABLE_ARG = "staging-table";
    public static final String CLEAR_STAGING_TABLE_ARG = "clear-staging-table";
    public static final String COLUMNS_ARG = "columns";
    public static final String SPLIT_BY_ARG = "split-by";
    public static final String SPLIT_LIMIT_ARG = "split-limit";
    public static final String WHERE_ARG = "where";
    public static final String HADOOP_HOME_ARG = "hadoop-home";
    public static final String HADOOP_MAPRED_HOME_ARG = "hadoop-mapred-home";
    public static final String HIVE_HOME_ARG = "hive-home";
    public static final String WAREHOUSE_DIR_ARG = "warehouse-dir";
    public static final String TARGET_DIR_ARG = "target-dir";
    public static final String APPEND_ARG = "append";
    public static final String DELETE_ARG = "delete-target-dir";
    public static final String NULL_STRING = "null-string";
    public static final String INPUT_NULL_STRING = "input-null-string";
    public static final String NULL_NON_STRING = "null-non-string";
    public static final String INPUT_NULL_NON_STRING = "input-null-non-string";
    public static final String MAP_COLUMN_JAVA = "map-column-java";
    public static final String MAP_COLUMN_HIVE = "map-column-hive";

    public static final String FMT_SEQUENCEFILE_ARG = "as-sequencefile";
    public static final String FMT_TEXTFILE_ARG = "as-textfile";
    public static final String FMT_AVRODATAFILE_ARG = "as-avrodatafile";
    public static final String FMT_PARQUETFILE_ARG = "as-parquetfile";
    public static final String HIVE_IMPORT_ARG = "hive-import";
    public static final String HIVE_TABLE_ARG = "hive-table";
    public static final String HIVE_DATABASE_ARG = "hive-database";
    public static final String HIVE_OVERWRITE_ARG = "hive-overwrite";
    public static final String HIVE_DROP_DELIMS_ARG = "hive-drop-import-delims";
    public static final String HIVE_DELIMS_REPLACEMENT_ARG =
        "hive-delims-replacement";
    public static final String HIVE_PARTITION_KEY_ARG = "hive-partition-key";
    public static final String HIVE_PARTITION_VALUE_ARG = "hive-partition-value";
    public static final String HCATCALOG_PARTITION_KEYS_ARG =
        "hcatalog-partition-keys";
    public static final String HCATALOG_PARTITION_VALUES_ARG =
        "hcatalog-partition-values";
    public static final String CREATE_HIVE_TABLE_ARG =
        "create-hive-table";
    public static final String HCATALOG_TABLE_ARG = "hcatalog-table";
    public static final String HCATALOG_DATABASE_ARG = "hcatalog-database";
    public static final String CREATE_HCATALOG_TABLE_ARG =
        "create-hcatalog-table";
    public static final String DROP_AND_CREATE_HCATALOG_TABLE =
        "drop-and-create-hcatalog-table";
    public static final String HCATALOG_STORAGE_STANZA_ARG =
        "hcatalog-storage-stanza";
    public static final String HCATALOG_HOME_ARG = "hcatalog-home";
    public static final String MAPREDUCE_JOB_NAME = "mapreduce-job-name";
    public static final String NUM_MAPPERS_ARG = "num-mappers";
    public static final String NUM_MAPPERS_SHORT_ARG = "m";
    public static final String COMPRESS_ARG = "compress";
    public static final String COMPRESSION_CODEC_ARG = "compression-codec";
    public static final String COMPRESS_SHORT_ARG = "z";
    public static final String DIRECT_SPLIT_SIZE_ARG = "direct-split-size";
    public static final String INLINE_LOB_LIMIT_ARG = "inline-lob-limit";
    public static final String FETCH_SIZE_ARG = "fetch-size";
    public static final String EXPORT_PATH_ARG = "export-dir";
    public static final String FIELDS_TERMINATED_BY_ARG = "fields-terminated-by";
    public static final String LINES_TERMINATED_BY_ARG = "lines-terminated-by";
    public static final String OPTIONALLY_ENCLOSED_BY_ARG =
        "optionally-enclosed-by";
    public static final String ENCLOSED_BY_ARG = "enclosed-by";
    public static final String ESCAPED_BY_ARG = "escaped-by";
    public static final String MYSQL_DELIMITERS_ARG = "mysql-delimiters";
    public static final String INPUT_FIELDS_TERMINATED_BY_ARG =
        "input-fields-terminated-by";
    public static final String INPUT_LINES_TERMINATED_BY_ARG =
        "input-lines-terminated-by";
    public static final String INPUT_OPTIONALLY_ENCLOSED_BY_ARG =
        "input-optionally-enclosed-by";
    public static final String INPUT_ENCLOSED_BY_ARG = "input-enclosed-by";
    public static final String INPUT_ESCAPED_BY_ARG = "input-escaped-by";
    public static final String CODE_OUT_DIR_ARG = "outdir";
    public static final String BIN_OUT_DIR_ARG = "bindir";
    public static final String PACKAGE_NAME_ARG = "package-name";
    public static final String CLASS_NAME_ARG = "class-name";
    public static final String JAR_FILE_NAME_ARG = "jar-file";
    public static final String SQL_QUERY_ARG = "query";
    public static final String SQL_QUERY_BOUNDARY = "boundary-query";
    public static final String SQL_QUERY_SHORT_ARG = "e";
    public static final String VERBOSE_ARG = "verbose";
    public static final String HELP_ARG = "help";
    public static final String TEMP_ROOTDIR_ARG = "temporary-rootdir";
    public static final String UPDATE_KEY_ARG = "update-key";
    public static final String UPDATE_MODE_ARG = "update-mode";
    public static final String CALL_ARG = "call";
    public static final String SKIP_DISTCACHE_ARG = "skip-dist-cache";
    public static final String RELAXED_ISOLATION = "relaxed-isolation";

    // Arguments for validation.
    public static final String VALIDATE_ARG = "validate";
    public static final String VALIDATOR_CLASS_ARG = "validator";
    public static final String VALIDATION_THRESHOLD_CLASS_ARG =
        "validation-threshold";
    public static final String VALIDATION_FAILURE_HANDLER_CLASS_ARG =
        "validation-failurehandler";

    // Arguments for incremental imports.
    public static final String INCREMENT_TYPE_ARG = "incremental";
    public static final String INCREMENT_COL_ARG = "check-column";
    public static final String INCREMENT_LAST_VAL_ARG = "last-value";

    // Arguments for all table imports.
    public static final String ALL_TABLE_EXCLUDES_ARG = "exclude-tables";

    // HBase arguments.
    public static final String HBASE_TABLE_ARG = "hbase-table";
    public static final String HBASE_COL_FAM_ARG = "column-family";
    public static final String HBASE_ROW_KEY_ARG = "hbase-row-key";
    public static final String HBASE_BULK_LOAD_ENABLED_ARG =
        "hbase-bulkload";
    public static final String HBASE_CREATE_TABLE_ARG = "hbase-create-table";

    // Accumulo arguments.
    public static final String ACCUMULO_TABLE_ARG = "accumulo-table";
    public static final String ACCUMULO_COL_FAM_ARG = "accumulo-column-family";
    public static final String ACCUMULO_ROW_KEY_ARG = "accumulo-row-key";
    public static final String ACCUMULO_VISIBILITY_ARG = "accumulo-visibility";
    public static final String ACCUMULO_CREATE_TABLE_ARG
        = "accumulo-create-table";
    public static final String ACCUMULO_BATCH_SIZE_ARG = "accumulo-batch-size";
    public static final String ACCUMULO_MAX_LATENCY_ARG = "accumulo-max-latency";
    public static final String ACCUMULO_ZOOKEEPERS_ARG = "accumulo-zookeepers";
    public static final String ACCUMULO_INSTANCE_ARG = "accumulo-instance";
    public static final String ACCUMULO_USER_ARG = "accumulo-user";
    public static final String ACCUMULO_PASSWORD_ARG = "accumulo-password";

    // Arguments for the saved job management system.
    public static final String STORAGE_METASTORE_ARG = "meta-connect";
    public static final String JOB_CMD_CREATE_ARG = "create";
    public static final String JOB_CMD_DELETE_ARG = "delete";
    public static final String JOB_CMD_EXEC_ARG = "exec";
    public static final String JOB_CMD_LIST_ARG = "list";
    public static final String JOB_CMD_SHOW_ARG = "show";

    // Arguments for the metastore.
    public static final String METASTORE_SHUTDOWN_ARG = "shutdown";

    // Arguments for merging datasets.
    public static final String NEW_DATASET_ARG = "new-data";
    public static final String OLD_DATASET_ARG = "onto";
    public static final String MERGE_KEY_ARG = "merge-key";

    // Reset number of mappers to one if there is no primary key avaliable and
    // split by column is explicitly not provided

    public static final String AUTORESET_TO_ONE_MAPPER = "autoreset-to-one-mapper";

    public BaseSqoopTool() {
    }

    public BaseSqoopTool(String toolName) {
    }

    /**
     * Examines a subset of the arrray presented, and determines if it contains any non-empty arguments. If so, logs the
     * arguments and returns true.
     *
     * @param argv   an array of strings to check.
     * @param offset the first element of the array to check
     * @param len    the number of elements to check
     * @return true if there are any non-null, non-empty argument strings present.
     */
    protected boolean hasUnrecognizedArgs(String[] argv, int offset, int len) {
        if (argv == null) {
            return false;
        }

        boolean unrecognized = false;
        boolean printedBanner = false;
        for (int i = offset; i < Math.min(argv.length, offset + len); i++) {
            if (argv[i] != null && argv[i].length() > 0) {
                if (!printedBanner) {
                    LOG.error("Error parsing arguments for " + ":");
                    printedBanner = true;
                }
                LOG.error("Unrecognized argument: " + argv[i]);
                unrecognized = true;
            }
        }

        return unrecognized;
    }

    protected boolean hasUnrecognizedArgs(String[] argv) {
        if (null == argv) {
            return false;
        }
        return hasUnrecognizedArgs(argv, 0, argv.length);
    }

    /**
     * If argv contains an entry "--", return an array containing all elements after the "--" separator. Otherwise,
     * return null.
     *
     * @param argv a set of arguments to scan for the subcommand arguments.
     */
    protected String[] getSubcommandArgs(String[] argv) {
        if (null == argv) {
            return null;
        }

        for (int i = 0; i < argv.length; i++) {
            if (argv[i].equals("--")) {
                return Arrays.copyOfRange(argv, i + 1, argv.length);
            }
        }

        return null;
    }

    /**
     * @return RelatedOptions used by job management tools.
     */
    protected RelatedOptions getJobOptions() {
        RelatedOptions relatedOpts = new RelatedOptions(
            "Job management arguments");
        relatedOpts.addOption(OptionBuilder.withArgName("jdbc-uri")
            .hasArg()
            .withDescription("Specify JDBC connect string for the metastore")
            .withLongOpt(STORAGE_METASTORE_ARG)
            .create());

        // Create an option-group surrounding the operations a user
        // can perform on jobs.
        OptionGroup group = new OptionGroup();
        group.addOption(OptionBuilder.withArgName("job-id")
            .hasArg()
            .withDescription("Create a new saved job")
            .withLongOpt(JOB_CMD_CREATE_ARG)
            .create());
        group.addOption(OptionBuilder.withArgName("job-id")
            .hasArg()
            .withDescription("Delete a saved job")
            .withLongOpt(JOB_CMD_DELETE_ARG)
            .create());
        group.addOption(OptionBuilder.withArgName("job-id")
            .hasArg()
            .withDescription("Show the parameters for a saved job")
            .withLongOpt(JOB_CMD_SHOW_ARG)
            .create());

        Option execOption = OptionBuilder.withArgName("job-id")
            .hasArg()
            .withDescription("Run a saved job")
            .withLongOpt(JOB_CMD_EXEC_ARG)
            .create();
        group.addOption(execOption);

        group.addOption(OptionBuilder
            .withDescription("List saved jobs")
            .withLongOpt(JOB_CMD_LIST_ARG)
            .create());

        relatedOpts.addOptionGroup(group);

        // Since the "common" options aren't used in the job tool,
        // add these settings here.
        relatedOpts.addOption(OptionBuilder
            .withDescription("Print more information while working")
            .withLongOpt(VERBOSE_ARG)
            .create());
        relatedOpts.addOption(OptionBuilder
            .withDescription("Print usage instructions")
            .withLongOpt(HELP_ARG)
            .create());

        return relatedOpts;
    }

    /**
     * @return RelatedOptions used by most/all Sqoop tools.
     */
    protected RelatedOptions getCommonOptions() {
        // Connection args (common)
        RelatedOptions commonOpts = new RelatedOptions("Common arguments");
        commonOpts.addOption(OptionBuilder.withArgName("jdbc-uri")
            .hasArg().withDescription("Specify JDBC connect string")
            .withLongOpt(CONNECT_STRING_ARG)
            .create());
        commonOpts.addOption(OptionBuilder.withArgName("class-name")
            .hasArg().withDescription("Specify connection manager class name")
            .withLongOpt(CONN_MANAGER_CLASS_NAME)
            .create());
        commonOpts.addOption(OptionBuilder.withArgName("properties-file")
            .hasArg().withDescription("Specify connection parameters file")
            .withLongOpt(CONNECT_PARAM_FILE)
            .create());
        commonOpts.addOption(OptionBuilder.withArgName("class-name")
            .hasArg().withDescription("Manually specify JDBC driver class to use")
            .withLongOpt(DRIVER_ARG)
            .create());
        commonOpts.addOption(OptionBuilder.withArgName("username")
            .hasArg().withDescription("Set authentication username")
            .withLongOpt(USERNAME_ARG)
            .create());
        commonOpts.addOption(OptionBuilder.withArgName("password")
            .hasArg().withDescription("Set authentication password")
            .withLongOpt(PASSWORD_ARG)
            .create());
        commonOpts.addOption(OptionBuilder.withArgName(PASSWORD_PATH_ARG)
            .hasArg().withDescription("Set authentication password file path")
            .withLongOpt(PASSWORD_PATH_ARG)
            .create());
        commonOpts.addOption(OptionBuilder
            .withDescription("Read password from console")
            .create(PASSWORD_PROMPT_ARG));
        commonOpts.addOption(OptionBuilder.withArgName(PASSWORD_ALIAS_ARG)
            .hasArg().withDescription("Credential provider password alias")
            .withLongOpt(PASSWORD_ALIAS_ARG)
            .create());
        commonOpts.addOption(OptionBuilder.withArgName("dir")
            .hasArg().withDescription("Override $HADOOP_MAPRED_HOME_ARG")
            .withLongOpt(HADOOP_MAPRED_HOME_ARG)
            .create());

        commonOpts.addOption(OptionBuilder.withArgName("hdir")
            .hasArg().withDescription("Override $HADOOP_MAPRED_HOME_ARG")
            .withLongOpt(HADOOP_HOME_ARG)
            .create());
        commonOpts.addOption(OptionBuilder
            .withDescription("Skip copying jars to distributed cache")
            .withLongOpt(SKIP_DISTCACHE_ARG)
            .create());

        // misc (common)
        commonOpts.addOption(OptionBuilder
            .withDescription("Print more information while working")
            .withLongOpt(VERBOSE_ARG)
            .create());
        commonOpts.addOption(OptionBuilder
            .withDescription("Print usage instructions")
            .withLongOpt(HELP_ARG)
            .create());
        commonOpts.addOption(OptionBuilder
            .withDescription("Defines the temporary root directory for the import")
            .withLongOpt(TEMP_ROOTDIR_ARG)
            .hasArg()
            .withArgName("rootdir")
            .create());
        // relax isolation requirements
        commonOpts.addOption(OptionBuilder
            .withDescription("Use read-uncommitted isolation for imports")
            .withLongOpt(RELAXED_ISOLATION)
            .create());

        return commonOpts;
    }

    /**
     * @param explicitHiveImport true if the user has an explicit --hive-import available, or false if this is implied
     *                           by the tool.
     * @return options governing interaction with Hive
     */
    protected RelatedOptions getHiveOptions(boolean explicitHiveImport) {
        RelatedOptions hiveOpts = new RelatedOptions("Hive arguments");
        if (explicitHiveImport) {
            hiveOpts.addOption(OptionBuilder
                .withDescription("Import tables into Hive "
                    + "(Uses Hive's default delimiters if none are set.)")
                .withLongOpt(HIVE_IMPORT_ARG)
                .create());
        }

        hiveOpts.addOption(OptionBuilder.withArgName("dir")
            .hasArg().withDescription("Override $HIVE_HOME")
            .withLongOpt(HIVE_HOME_ARG)
            .create());
        hiveOpts.addOption(OptionBuilder
            .withDescription("Overwrite existing data in the Hive table")
            .withLongOpt(HIVE_OVERWRITE_ARG)
            .create());
        hiveOpts.addOption(OptionBuilder
            .withDescription("Fail if the target hive table exists")
            .withLongOpt(CREATE_HIVE_TABLE_ARG)
            .create());
        hiveOpts.addOption(OptionBuilder.withArgName("table-name")
            .hasArg()
            .withDescription("Sets the table name to use when importing to hive")
            .withLongOpt(HIVE_TABLE_ARG)
            .create());
        hiveOpts.addOption(OptionBuilder.withArgName("database-name")
            .hasArg()
            .withDescription("Sets the database name to use when importing to hive")
            .withLongOpt(HIVE_DATABASE_ARG)
            .create());
        hiveOpts.addOption(OptionBuilder
            .withDescription("Drop Hive record \\0x01 and row delimiters "
                + "(\\n\\r) from imported string fields")
            .withLongOpt(HIVE_DROP_DELIMS_ARG)
            .create());
        hiveOpts.addOption(OptionBuilder
            .hasArg()
            .withDescription("Replace Hive record \\0x01 and row delimiters "
                + "(\\n\\r) from imported string fields with user-defined string")
            .withLongOpt(HIVE_DELIMS_REPLACEMENT_ARG)
            .create());
        hiveOpts.addOption(OptionBuilder.withArgName("partition-key")
            .hasArg()
            .withDescription("Sets the partition key to use when importing to hive")
            .withLongOpt(HIVE_PARTITION_KEY_ARG)
            .create());
        hiveOpts.addOption(OptionBuilder.withArgName("partition-value")
            .hasArg()
            .withDescription("Sets the partition value to use when importing "
                + "to hive")
            .withLongOpt(HIVE_PARTITION_VALUE_ARG)
            .create());
        hiveOpts.addOption(OptionBuilder
            .hasArg()
            .withDescription("Override mapping for specific column to hive"
                + " types.")
            .withLongOpt(MAP_COLUMN_HIVE)
            .create());

        return hiveOpts;
    }

    /**
     * @return options governing interaction with HCatalog.
     */
    protected RelatedOptions getHCatalogOptions() {
        RelatedOptions hCatOptions = new RelatedOptions("HCatalog arguments");
        hCatOptions.addOption(OptionBuilder
            .hasArg()
            .withDescription("HCatalog table name")
            .withLongOpt(HCATALOG_TABLE_ARG)
            .create());
        hCatOptions.addOption(OptionBuilder
            .hasArg()
            .withDescription("HCatalog database name")
            .withLongOpt(HCATALOG_DATABASE_ARG)
            .create());

        hCatOptions.addOption(OptionBuilder.withArgName("dir")
            .hasArg().withDescription("Override $HIVE_HOME")
            .withLongOpt(HIVE_HOME_ARG)
            .create());
        hCatOptions.addOption(OptionBuilder.withArgName("hdir")
            .hasArg().withDescription("Override $HCAT_HOME")
            .withLongOpt(HCATALOG_HOME_ARG)
            .create());
        hCatOptions.addOption(OptionBuilder.withArgName("partition-key")
            .hasArg()
            .withDescription("Sets the partition key to use when importing to hive")
            .withLongOpt(HIVE_PARTITION_KEY_ARG)
            .create());
        hCatOptions.addOption(OptionBuilder.withArgName("partition-value")
            .hasArg()
            .withDescription("Sets the partition value to use when importing "
                + "to hive")
            .withLongOpt(HIVE_PARTITION_VALUE_ARG)
            .create());
        hCatOptions.addOption(OptionBuilder
            .hasArg()
            .withDescription("Override mapping for specific column to hive"
                + " types.")
            .withLongOpt(MAP_COLUMN_HIVE)
            .create());
        hCatOptions.addOption(OptionBuilder.withArgName("partition-key")
            .hasArg()
            .withDescription("Sets the partition keys to use when importing to hive")
            .withLongOpt(HCATCALOG_PARTITION_KEYS_ARG)
            .create());
        hCatOptions.addOption(OptionBuilder.withArgName("partition-value")
            .hasArg()
            .withDescription("Sets the partition values to use when importing "
                + "to hive")
            .withLongOpt(HCATALOG_PARTITION_VALUES_ARG)
            .create());
        return hCatOptions;
    }

    protected RelatedOptions getHCatImportOnlyOptions() {
        RelatedOptions hCatOptions = new RelatedOptions(
            "HCatalog import specific options");
        hCatOptions.addOption(OptionBuilder
            .withDescription("Create HCatalog before import")
            .withLongOpt(CREATE_HCATALOG_TABLE_ARG)
            .create());
        hCatOptions.addOption(OptionBuilder
            .withDescription("Drop and Create HCatalog before import")
            .withLongOpt(DROP_AND_CREATE_HCATALOG_TABLE)
            .create());
        hCatOptions.addOption(OptionBuilder
            .hasArg()
            .withDescription("HCatalog storage stanza for table creation")
            .withLongOpt(HCATALOG_STORAGE_STANZA_ARG)
            .create());
        return hCatOptions;
    }

    /**
     * @return options governing output format delimiters
     */
    protected RelatedOptions getOutputFormatOptions() {
        RelatedOptions formatOpts = new RelatedOptions(
            "Output line formatting arguments");
        formatOpts.addOption(OptionBuilder.withArgName("char")
            .hasArg()
            .withDescription("Sets the field separator character")
            .withLongOpt(FIELDS_TERMINATED_BY_ARG)
            .create());
        formatOpts.addOption(OptionBuilder.withArgName("char")
            .hasArg()
            .withDescription("Sets the end-of-line character")
            .withLongOpt(LINES_TERMINATED_BY_ARG)
            .create());
        formatOpts.addOption(OptionBuilder.withArgName("char")
            .hasArg()
            .withDescription("Sets a field enclosing character")
            .withLongOpt(OPTIONALLY_ENCLOSED_BY_ARG)
            .create());
        formatOpts.addOption(OptionBuilder.withArgName("char")
            .hasArg()
            .withDescription("Sets a required field enclosing character")
            .withLongOpt(ENCLOSED_BY_ARG)
            .create());
        formatOpts.addOption(OptionBuilder.withArgName("char")
            .hasArg()
            .withDescription("Sets the escape character")
            .withLongOpt(ESCAPED_BY_ARG)
            .create());
        formatOpts.addOption(OptionBuilder
            .withDescription("Uses MySQL's default delimiter set: "
                + "fields: ,  lines: \\n  escaped-by: \\  optionally-enclosed-by: '")
            .withLongOpt(MYSQL_DELIMITERS_ARG)
            .create());

        return formatOpts;
    }

    /**
     * @return options governing input format delimiters.
     */
    protected RelatedOptions getInputFormatOptions() {
        RelatedOptions inputFormatOpts =
            new RelatedOptions("Input parsing arguments");
        inputFormatOpts.addOption(OptionBuilder.withArgName("char")
            .hasArg()
            .withDescription("Sets the input field separator")
            .withLongOpt(INPUT_FIELDS_TERMINATED_BY_ARG)
            .create());
        inputFormatOpts.addOption(OptionBuilder.withArgName("char")
            .hasArg()
            .withDescription("Sets the input end-of-line char")
            .withLongOpt(INPUT_LINES_TERMINATED_BY_ARG)
            .create());
        inputFormatOpts.addOption(OptionBuilder.withArgName("char")
            .hasArg()
            .withDescription("Sets a field enclosing character")
            .withLongOpt(INPUT_OPTIONALLY_ENCLOSED_BY_ARG)
            .create());
        inputFormatOpts.addOption(OptionBuilder.withArgName("char")
            .hasArg()
            .withDescription("Sets a required field encloser")
            .withLongOpt(INPUT_ENCLOSED_BY_ARG)
            .create());
        inputFormatOpts.addOption(OptionBuilder.withArgName("char")
            .hasArg()
            .withDescription("Sets the input escape character")
            .withLongOpt(INPUT_ESCAPED_BY_ARG)
            .create());

        return inputFormatOpts;
    }

    /**
     * @param multiTable true if these options will be used for bulk code-gen.
     * @return options related to code generation.
     */
    protected RelatedOptions getCodeGenOpts(boolean multiTable) {
        RelatedOptions codeGenOpts =
            new RelatedOptions("Code generation arguments");
        codeGenOpts.addOption(OptionBuilder.withArgName("dir")
            .hasArg()
            .withDescription("Output directory for generated code")
            .withLongOpt(CODE_OUT_DIR_ARG)
            .create());
        codeGenOpts.addOption(OptionBuilder.withArgName("dir")
            .hasArg()
            .withDescription("Output directory for compiled objects")
            .withLongOpt(BIN_OUT_DIR_ARG)
            .create());
        codeGenOpts.addOption(OptionBuilder.withArgName("name")
            .hasArg()
            .withDescription("Put auto-generated classes in this package")
            .withLongOpt(PACKAGE_NAME_ARG)
            .create());
        codeGenOpts.addOption(OptionBuilder.withArgName("null-str")
            .hasArg()
            .withDescription("Null string representation")
            .withLongOpt(NULL_STRING)
            .create());
        codeGenOpts.addOption(OptionBuilder.withArgName("null-str")
            .hasArg()
            .withDescription("Input null string representation")
            .withLongOpt(INPUT_NULL_STRING)
            .create());
        codeGenOpts.addOption(OptionBuilder.withArgName("null-str")
            .hasArg()
            .withDescription("Null non-string representation")
            .withLongOpt(NULL_NON_STRING)
            .create());
        codeGenOpts.addOption(OptionBuilder.withArgName("null-str")
            .hasArg()
            .withDescription("Input null non-string representation")
            .withLongOpt(INPUT_NULL_NON_STRING)
            .create());
        codeGenOpts.addOption(OptionBuilder
            .hasArg()
            .withDescription("Override mapping for specific columns to java types")
            .withLongOpt(MAP_COLUMN_JAVA)
            .create());

        if (!multiTable) {
            codeGenOpts.addOption(OptionBuilder.withArgName("name")
                .hasArg()
                .withDescription("Sets the generated class name. "
                    + "This overrides --" + PACKAGE_NAME_ARG + ". When combined "
                    + "with --" + JAR_FILE_NAME_ARG + ", sets the input class.")
                .withLongOpt(CLASS_NAME_ARG)
                .create());
        }
        return codeGenOpts;
    }

    protected RelatedOptions getHBaseOptions() {
        RelatedOptions hbaseOpts =
            new RelatedOptions("HBase arguments");
        hbaseOpts.addOption(OptionBuilder.withArgName("table")
            .hasArg()
            .withDescription("Import to <table> in HBase")
            .withLongOpt(HBASE_TABLE_ARG)
            .create());
        hbaseOpts.addOption(OptionBuilder.withArgName("family")
            .hasArg()
            .withDescription("Sets the target column family for the import")
            .withLongOpt(HBASE_COL_FAM_ARG)
            .create());
        hbaseOpts.addOption(OptionBuilder.withArgName("col")
            .hasArg()
            .withDescription("Specifies which input column to use as the row key")
            .withLongOpt(HBASE_ROW_KEY_ARG)
            .create());
        hbaseOpts.addOption(OptionBuilder
            .withDescription("Enables HBase bulk loading")
            .withLongOpt(HBASE_BULK_LOAD_ENABLED_ARG)
            .create());
        hbaseOpts.addOption(OptionBuilder
            .withDescription("If specified, create missing HBase tables")
            .withLongOpt(HBASE_CREATE_TABLE_ARG)
            .create());

        return hbaseOpts;
    }

    protected RelatedOptions getAccumuloOptions() {
        RelatedOptions accumuloOpts =
            new RelatedOptions("Accumulo arguments");
        accumuloOpts.addOption(OptionBuilder.withArgName("table")
            .hasArg()
            .withDescription("Import to <table> in Accumulo")
            .withLongOpt(ACCUMULO_TABLE_ARG)
            .create());
        accumuloOpts.addOption(OptionBuilder.withArgName("family")
            .hasArg()
            .withDescription("Sets the target column family for the import")
            .withLongOpt(ACCUMULO_COL_FAM_ARG)
            .create());
        accumuloOpts.addOption(OptionBuilder.withArgName("col")
            .hasArg()
            .withDescription("Specifies which input column to use as the row key")
            .withLongOpt(ACCUMULO_ROW_KEY_ARG)
            .create());
        accumuloOpts.addOption(OptionBuilder.withArgName("vis")
            .hasArg()
            .withDescription("Visibility token to be applied to all rows imported")
            .withLongOpt(ACCUMULO_VISIBILITY_ARG)
            .create());
        accumuloOpts.addOption(OptionBuilder
            .withDescription("If specified, create missing Accumulo tables")
            .withLongOpt(ACCUMULO_CREATE_TABLE_ARG)
            .create());
        accumuloOpts.addOption(OptionBuilder.withArgName("size")
            .hasArg()
            .withDescription("Batch size in bytes")
            .withLongOpt(ACCUMULO_BATCH_SIZE_ARG)
            .create());
        accumuloOpts.addOption(OptionBuilder.withArgName("latency")
            .hasArg()
            .withDescription("Max write latency in milliseconds")
            .withLongOpt(ACCUMULO_MAX_LATENCY_ARG)
            .create());
        accumuloOpts.addOption(OptionBuilder.withArgName("zookeepers")
            .hasArg()
            .withDescription("Comma-separated list of zookeepers (host:port)")
            .withLongOpt(ACCUMULO_ZOOKEEPERS_ARG)
            .create());
        accumuloOpts.addOption(OptionBuilder.withArgName("instance")
            .hasArg()
            .withDescription("Accumulo instance name.")
            .withLongOpt(ACCUMULO_INSTANCE_ARG)
            .create());
        accumuloOpts.addOption(OptionBuilder.withArgName("user")
            .hasArg()
            .withDescription("Accumulo user name.")
            .withLongOpt(ACCUMULO_USER_ARG)
            .create());
        accumuloOpts.addOption(OptionBuilder.withArgName("password")
            .hasArg()
            .withDescription("Accumulo password.")
            .withLongOpt(ACCUMULO_PASSWORD_ARG)
            .create());

        return accumuloOpts;
    }

    protected void applyAccumuloOptions(CommandLine in, SqoopOptions out) {
        if (in.hasOption(ACCUMULO_TABLE_ARG)) {
            out.setAccumuloTable(in.getOptionValue(ACCUMULO_TABLE_ARG));
        }

        if (in.hasOption(ACCUMULO_COL_FAM_ARG)) {
            out.setAccumuloColFamily(in.getOptionValue(ACCUMULO_COL_FAM_ARG));
        }

        if (in.hasOption(ACCUMULO_ROW_KEY_ARG)) {
            out.setAccumuloRowKeyColumn(in.getOptionValue(ACCUMULO_ROW_KEY_ARG));
        }

        if (in.hasOption(ACCUMULO_VISIBILITY_ARG)) {
            out.setAccumuloVisibility(in.getOptionValue(ACCUMULO_VISIBILITY_ARG));
        }

        if (in.hasOption(ACCUMULO_CREATE_TABLE_ARG)) {
            out.setCreateAccumuloTable(true);
        }

        if (in.hasOption(ACCUMULO_BATCH_SIZE_ARG)) {
            out.setAccumuloBatchSize(Long.parseLong(
                in.getOptionValue(ACCUMULO_BATCH_SIZE_ARG)));
        }

        if (in.hasOption(ACCUMULO_MAX_LATENCY_ARG)) {
            out.setAccumuloMaxLatency(Long.parseLong(
                in.getOptionValue(ACCUMULO_MAX_LATENCY_ARG)));
        }

        if (in.hasOption(ACCUMULO_ZOOKEEPERS_ARG)) {
            out.setAccumuloZookeepers(in.getOptionValue(ACCUMULO_ZOOKEEPERS_ARG));
        }

        if (in.hasOption(ACCUMULO_INSTANCE_ARG)) {
            out.setAccumuloInstance(in.getOptionValue(ACCUMULO_INSTANCE_ARG));
        }

        if (in.hasOption(ACCUMULO_USER_ARG)) {
            out.setAccumuloUser(in.getOptionValue(ACCUMULO_USER_ARG));
        }

        if (in.hasOption(ACCUMULO_PASSWORD_ARG)) {
            out.setAccumuloPassword(in.getOptionValue(ACCUMULO_PASSWORD_ARG));
        }
    }

    @SuppressWarnings("static-access")
    protected void addValidationOpts(RelatedOptions validationOptions) {
        validationOptions.addOption(OptionBuilder
            .withDescription("Validate the copy using the configured validator")
            .withLongOpt(VALIDATE_ARG)
            .create());
        validationOptions.addOption(OptionBuilder
            .withArgName(VALIDATOR_CLASS_ARG).hasArg()
            .withDescription("Fully qualified class name for the Validator")
            .withLongOpt(VALIDATOR_CLASS_ARG)
            .create());
        validationOptions.addOption(OptionBuilder
            .withArgName(VALIDATION_THRESHOLD_CLASS_ARG).hasArg()
            .withDescription("Fully qualified class name for ValidationThreshold")
            .withLongOpt(VALIDATION_THRESHOLD_CLASS_ARG)
            .create());
        validationOptions.addOption(OptionBuilder
            .withArgName(VALIDATION_FAILURE_HANDLER_CLASS_ARG).hasArg()
            .withDescription("Fully qualified class name for "
                + "ValidationFailureHandler")
            .withLongOpt(VALIDATION_FAILURE_HANDLER_CLASS_ARG)
            .create());
    }

    public void printHelp(ToolOptions opts) {
        System.out.println("usage: sqoop "
            + " [GENERIC-ARGS] [TOOL-ARGS]");
        System.out.println("");

        opts.printHelp();

        System.out.println("");
        System.out.println("Generic Hadoop command-line arguments:");
        System.out.println("(must preceed any tool-specific arguments)");
        ToolRunner.printGenericCommandUsage(System.out);
    }

    /**
     * Apply common command-line to the state.
     */
    protected void applyCommonOptions(CommandLine in, SqoopOptions out)
        throws InvalidOptionsException {

        // common options.
        if (in.hasOption(VERBOSE_ARG)) {
            // Immediately switch into DEBUG logging.
            out.setVerbose(true);
            LOG.debug("Enabled debug logging.");
        }

        if (in.hasOption(HELP_ARG)) {
            ToolOptions toolOpts = new ToolOptions();
            configureOptions(toolOpts);
            printHelp(toolOpts);
            throw new InvalidOptionsException("");
        }

        if (in.hasOption(TEMP_ROOTDIR_ARG)) {
            out.setTempRootDir(in.getOptionValue(TEMP_ROOTDIR_ARG));
        }

        if (in.hasOption(CONNECT_STRING_ARG)) {
            out.setConnectString(in.getOptionValue(CONNECT_STRING_ARG));
        }

        if (in.hasOption(CONN_MANAGER_CLASS_NAME)) {
            out.setConnManagerClassName(in.getOptionValue(CONN_MANAGER_CLASS_NAME));
        }

        if (in.hasOption(CONNECT_PARAM_FILE)) {
            File paramFile = new File(in.getOptionValue(CONNECT_PARAM_FILE));
            if (!paramFile.exists()) {
                throw new InvalidOptionsException(
                    "Specified connection parameter file not found: " + paramFile);
            }
            InputStream inStream = null;
            Properties connectionParams = new Properties();
            try {
                inStream = new FileInputStream(
                    new File(in.getOptionValue(CONNECT_PARAM_FILE)));
                connectionParams.load(inStream);
            } catch (IOException ex) {
                LOG.warn("Failed to load connection parameter file", ex);
                throw new InvalidOptionsException(
                    "Error while loading connection parameter file: "
                        + ex.getMessage());
            } finally {
                if (inStream != null) {
                    try {
                        inStream.close();
                    } catch (IOException ex) {
                        LOG.warn("Failed to close input stream", ex);
                    }
                }
            }
            LOG.debug("Loaded connection parameters: " + connectionParams);
            out.setConnectionParams(connectionParams);
        }

        if (in.hasOption(NULL_STRING)) {
            out.setNullStringValue(in.getOptionValue(NULL_STRING));
        }

        if (in.hasOption(INPUT_NULL_STRING)) {
            out.setInNullStringValue(in.getOptionValue(INPUT_NULL_STRING));
        }

        if (in.hasOption(NULL_NON_STRING)) {
            out.setNullNonStringValue(in.getOptionValue(NULL_NON_STRING));
        }

        if (in.hasOption(INPUT_NULL_NON_STRING)) {
            out.setInNullNonStringValue(in.getOptionValue(INPUT_NULL_NON_STRING));
        }

        if (in.hasOption(DRIVER_ARG)) {
            out.setDriverClassName(in.getOptionValue(DRIVER_ARG));
        }

        if (in.hasOption(SKIP_DISTCACHE_ARG)) {
            LOG.debug("Disabling dist cache");
            out.setSkipDistCache(true);
        }

        applyCredentialsOptions(in, out);

        if (in.hasOption(HADOOP_MAPRED_HOME_ARG)) {
            out.setHadoopMapRedHome(in.getOptionValue(HADOOP_MAPRED_HOME_ARG));
            // Only consider HADOOP_HOME if HADOOP_MAPRED_HOME is not set
        } else if (in.hasOption(HADOOP_HOME_ARG)) {
            out.setHadoopMapRedHome(in.getOptionValue(HADOOP_HOME_ARG));
        }
        if (in.hasOption(RELAXED_ISOLATION)) {
            out.setRelaxedIsolation(true);
        }
    }

    private void applyCredentialsOptions(CommandLine in, SqoopOptions out)
        throws InvalidOptionsException {
        if (in.hasOption(USERNAME_ARG)) {
            out.setUsername(in.getOptionValue(USERNAME_ARG));
            if (null == out.getPassword()) {
                // Set password to empty if the username is set first,
                // to ensure that they're either both null or neither is.
                out.setPassword("");
            }
        }

        if (in.hasOption(PASSWORD_ARG)) {
            LOG.warn("Setting your password on the command-line is insecure. "
                + "Consider using -" + PASSWORD_PROMPT_ARG + " instead.");
            out.setPassword(in.getOptionValue(PASSWORD_ARG));
        }

        if (in.hasOption(PASSWORD_PROMPT_ARG)) {
            out.setPasswordFromConsole();
        }

        if (in.hasOption(PASSWORD_PATH_ARG)) {
            if (in.hasOption(PASSWORD_ARG) || in.hasOption(PASSWORD_PROMPT_ARG)
                || in.hasOption(PASSWORD_ALIAS_ARG)) {
                throw new InvalidOptionsException("Only one of password, password "
                    + "alias or path to a password file must be specified.");
            }

        }
        if (in.hasOption(PASSWORD_ALIAS_ARG)) {
            if (in.hasOption(PASSWORD_ARG) || in.hasOption(PASSWORD_PROMPT_ARG)
                || in.hasOption(PASSWORD_PATH_ARG)) {
                throw new InvalidOptionsException("Only one of password, password "
                    + "alias or path to a password file must be specified.");
            }
            out.setPasswordAlias(in.getOptionValue(PASSWORD_ALIAS_ARG));
        }
    }

    public void configureOptions(ToolOptions opts) {
        // Default implementation does nothing.
    }

    protected void applyHiveOptions(CommandLine in, SqoopOptions out)
        throws InvalidOptionsException {

        if (in.hasOption(HIVE_HOME_ARG)) {
            out.setHiveHome(in.getOptionValue(HIVE_HOME_ARG));
        }

        if (in.hasOption(HIVE_IMPORT_ARG)) {
            out.setHiveImport(true);
        }

        if (in.hasOption(HIVE_OVERWRITE_ARG)) {
            out.setOverwriteHiveTable(true);
        }

        if (in.hasOption(CREATE_HIVE_TABLE_ARG)) {
            out.setFailIfHiveTableExists(true);
        }

        if (in.hasOption(HIVE_TABLE_ARG)) {
            out.setHiveTableName(in.getOptionValue(HIVE_TABLE_ARG));
        }

        if (in.hasOption(HIVE_DATABASE_ARG)) {
            out.setHiveDatabaseName(in.getOptionValue(HIVE_DATABASE_ARG));
        }

        if (in.hasOption(HIVE_DROP_DELIMS_ARG)) {
            out.setHiveDropDelims(true);
        }

        if (in.hasOption(HIVE_DELIMS_REPLACEMENT_ARG)) {
            out.setHiveDelimsReplacement(
                in.getOptionValue(HIVE_DELIMS_REPLACEMENT_ARG));
        }

        if (in.hasOption(HIVE_PARTITION_KEY_ARG)) {
            out.setHivePartitionKey(in.getOptionValue(HIVE_PARTITION_KEY_ARG));
        }

        if (in.hasOption(HIVE_PARTITION_VALUE_ARG)) {
            out.setHivePartitionValue(in.getOptionValue(HIVE_PARTITION_VALUE_ARG));
        }

        if (in.hasOption(MAP_COLUMN_HIVE)) {
            out.setMapColumnHive(in.getOptionValue(MAP_COLUMN_HIVE));
        }
    }

    protected void applyHCatalogOptions(CommandLine in, SqoopOptions out) {
        if (in.hasOption(HCATALOG_TABLE_ARG)) {
            out.setHCatTableName(in.getOptionValue(HCATALOG_TABLE_ARG));
        }

        if (in.hasOption(HCATALOG_DATABASE_ARG)) {
            out.setHCatDatabaseName(in.getOptionValue(HCATALOG_DATABASE_ARG));
        }

        if (in.hasOption(HCATALOG_STORAGE_STANZA_ARG)) {
            out.setHCatStorageStanza(in.getOptionValue(HCATALOG_STORAGE_STANZA_ARG));
        }

        if (in.hasOption(CREATE_HCATALOG_TABLE_ARG)) {
            out.setCreateHCatalogTable(true);
        }

        if (in.hasOption(DROP_AND_CREATE_HCATALOG_TABLE)) {
            out.setDropAndCreateHCatalogTable(true);
        }

        if (in.hasOption(HCATALOG_HOME_ARG)) {
            out.setHCatHome(in.getOptionValue(HCATALOG_HOME_ARG));
        }

        // Allow some of the hive options also

        if (in.hasOption(HIVE_HOME_ARG)) {
            out.setHiveHome(in.getOptionValue(HIVE_HOME_ARG));
        }

        if (in.hasOption(HCATCALOG_PARTITION_KEYS_ARG)) {
            out.setHCatalogPartitionKeys(
                in.getOptionValue(HCATCALOG_PARTITION_KEYS_ARG));
        }

        if (in.hasOption(HCATALOG_PARTITION_VALUES_ARG)) {
            out.setHCatalogPartitionValues(
                in.getOptionValue(HCATALOG_PARTITION_VALUES_ARG));
        }

        if (in.hasOption(HIVE_PARTITION_KEY_ARG)) {
            out.setHivePartitionKey(in.getOptionValue(HIVE_PARTITION_KEY_ARG));
        }

        if (in.hasOption(HIVE_PARTITION_VALUE_ARG)) {
            out.setHivePartitionValue(in.getOptionValue(HIVE_PARTITION_VALUE_ARG));
        }

        if (in.hasOption(MAP_COLUMN_HIVE)) {
            out.setMapColumnHive(in.getOptionValue(MAP_COLUMN_HIVE));
        }

    }

    protected void applyOutputFormatOptions(CommandLine in, SqoopOptions out)
        throws InvalidOptionsException,
        SqoopOptions.InvalidOptionsException {
        if (in.hasOption(FIELDS_TERMINATED_BY_ARG)) {
            out.setFieldsTerminatedBy(SqoopOptions.toChar(
                in.getOptionValue(FIELDS_TERMINATED_BY_ARG)));
            out.setExplicitOutputDelims(true);
        }

        if (in.hasOption(LINES_TERMINATED_BY_ARG)) {
            out.setLinesTerminatedBy(SqoopOptions.toChar(
                in.getOptionValue(LINES_TERMINATED_BY_ARG)));
            out.setExplicitOutputDelims(true);
        }

        if (in.hasOption(OPTIONALLY_ENCLOSED_BY_ARG)) {
            out.setEnclosedBy(SqoopOptions.toChar(
                in.getOptionValue(OPTIONALLY_ENCLOSED_BY_ARG)));
            out.setOutputEncloseRequired(false);
            out.setExplicitOutputDelims(true);
        }

        if (in.hasOption(ENCLOSED_BY_ARG)) {
            out.setEnclosedBy(SqoopOptions.toChar(
                in.getOptionValue(ENCLOSED_BY_ARG)));
            out.setOutputEncloseRequired(true);
            out.setExplicitOutputDelims(true);
        }

        if (in.hasOption(ESCAPED_BY_ARG)) {
            out.setEscapedBy(SqoopOptions.toChar(
                in.getOptionValue(ESCAPED_BY_ARG)));
            out.setExplicitOutputDelims(true);
        }

        if (in.hasOption(MYSQL_DELIMITERS_ARG)) {
            out.setOutputEncloseRequired(false);
            out.setFieldsTerminatedBy(',');
            out.setLinesTerminatedBy('\n');
            out.setEscapedBy('\\');
            out.setEnclosedBy('\'');
            out.setExplicitOutputDelims(true);
        }
    }

    protected void applyInputFormatOptions(CommandLine in, SqoopOptions out)
        throws InvalidOptionsException,
        SqoopOptions.InvalidOptionsException {
        if (in.hasOption(INPUT_FIELDS_TERMINATED_BY_ARG)) {
            out.setInputFieldsTerminatedBy(SqoopOptions.toChar(
                in.getOptionValue(INPUT_FIELDS_TERMINATED_BY_ARG)));
            out.setExplicitInputDelims(true);
        }

        if (in.hasOption(INPUT_LINES_TERMINATED_BY_ARG)) {
            out.setInputLinesTerminatedBy(SqoopOptions.toChar(
                in.getOptionValue(INPUT_LINES_TERMINATED_BY_ARG)));
            out.setExplicitInputDelims(true);
        }

        if (in.hasOption(INPUT_OPTIONALLY_ENCLOSED_BY_ARG)) {
            out.setInputEnclosedBy(SqoopOptions.toChar(
                in.getOptionValue(INPUT_OPTIONALLY_ENCLOSED_BY_ARG)));
            out.setInputEncloseRequired(false);
            out.setExplicitInputDelims(true);
        }

        if (in.hasOption(INPUT_ENCLOSED_BY_ARG)) {
            out.setInputEnclosedBy(SqoopOptions.toChar(
                in.getOptionValue(INPUT_ENCLOSED_BY_ARG)));
            out.setInputEncloseRequired(true);
            out.setExplicitInputDelims(true);
        }

        if (in.hasOption(INPUT_ESCAPED_BY_ARG)) {
            out.setInputEscapedBy(SqoopOptions.toChar(
                in.getOptionValue(INPUT_ESCAPED_BY_ARG)));
            out.setExplicitInputDelims(true);
        }
    }

    protected void applyCodeGenOptions(CommandLine in, SqoopOptions out,
                                       boolean multiTable) throws InvalidOptionsException {
        if (in.hasOption(CODE_OUT_DIR_ARG)) {
            out.setCodeOutputDir(in.getOptionValue(CODE_OUT_DIR_ARG));
        }

        if (in.hasOption(BIN_OUT_DIR_ARG)) {
            out.setJarOutputDir(in.getOptionValue(BIN_OUT_DIR_ARG));
        }

        if (in.hasOption(PACKAGE_NAME_ARG)) {
            out.setPackageName(in.getOptionValue(PACKAGE_NAME_ARG));
        }

        if (in.hasOption(MAP_COLUMN_JAVA)) {
            out.setMapColumnJava(in.getOptionValue(MAP_COLUMN_JAVA));
        }

        if (!multiTable && in.hasOption(CLASS_NAME_ARG)) {
            out.setClassName(in.getOptionValue(CLASS_NAME_ARG));
        }
    }

    protected void applyHBaseOptions(CommandLine in, SqoopOptions out) {
        if (in.hasOption(HBASE_TABLE_ARG)) {
            out.setHBaseTable(in.getOptionValue(HBASE_TABLE_ARG));
        }

        if (in.hasOption(HBASE_COL_FAM_ARG)) {
            out.setHBaseColFamily(in.getOptionValue(HBASE_COL_FAM_ARG));
        }

        if (in.hasOption(HBASE_ROW_KEY_ARG)) {
            out.setHBaseRowKeyColumn(in.getOptionValue(HBASE_ROW_KEY_ARG));
        }

        out.setHBaseBulkLoadEnabled(in.hasOption(HBASE_BULK_LOAD_ENABLED_ARG));

        if (in.hasOption(HBASE_CREATE_TABLE_ARG)) {
            out.setCreateHBaseTable(true);
        }
    }

    protected void applyValidationOptions(CommandLine in, SqoopOptions out)
        throws InvalidOptionsException {
        if (in.hasOption(VALIDATE_ARG)) {
            out.setValidationEnabled(true);
        }

        // Class Names are converted to Class in light of failing early
        if (in.hasOption(VALIDATOR_CLASS_ARG)) {
            out.setValidatorClass(
                getClassByName(in.getOptionValue(VALIDATOR_CLASS_ARG)));
        }

        if (in.hasOption(VALIDATION_THRESHOLD_CLASS_ARG)) {
            out.setValidationThresholdClass(
                getClassByName(in.getOptionValue(VALIDATION_THRESHOLD_CLASS_ARG)));
        }

        if (in.hasOption(VALIDATION_FAILURE_HANDLER_CLASS_ARG)) {
            out.setValidationFailureHandlerClass(getClassByName(
                in.getOptionValue(VALIDATION_FAILURE_HANDLER_CLASS_ARG)));
        }
    }

    protected Class<?> getClassByName(String className)
        throws InvalidOptionsException {
        try {
            return Class.forName(className, true,
                Thread.currentThread().getContextClassLoader());
        } catch (ClassNotFoundException e) {
            throw new InvalidOptionsException(e.getMessage());
        }
    }

    protected void validateCommonOptions(SqoopOptions options)
        throws InvalidOptionsException {
        if (options.getConnectString() == null) {
            throw new InvalidOptionsException(
                "Error: Required argument --connect is missing."
                    + HELP_STR);
        }
    }

    protected void validateCodeGenOptions(SqoopOptions options)
        throws InvalidOptionsException {
        if (options.getClassName() != null && options.getPackageName() != null) {
            throw new InvalidOptionsException(
                "--class-name overrides --package-name. You cannot use both."
                    + HELP_STR);
        }
    }

    protected void validateOutputFormatOptions(SqoopOptions options)
        throws InvalidOptionsException {
        if (options.doHiveImport()) {
            if (!options.explicitOutputDelims()) {
                // user hasn't manually specified delimiters, and wants to import
                // straight to Hive. Use Hive-style delimiters.
                LOG.info("Using Hive-specific delimiters for output. You can override");
                LOG.info("delimiters with --fields-terminated-by, etc.");
                options.setOutputDelimiters(DelimiterSet.HIVE_DELIMITERS);
            }

            if (options.getOutputEscapedBy() != DelimiterSet.NULL_CHAR) {
                LOG.warn("Hive does not support escape characters in fields;");
                LOG.warn("parse errors in Hive may result from using --escaped-by.");
            }

            if (options.getOutputEnclosedBy() != DelimiterSet.NULL_CHAR) {
                LOG.warn("Hive does not support quoted strings; parse errors");
                LOG.warn("in Hive may result from using --enclosed-by.");
            }
        }
    }

    protected void validateHiveOptions(SqoopOptions options)
        throws InvalidOptionsException {
        if (options.getHiveDelimsReplacement() != null
            && options.doHiveDropDelims()) {
            throw new InvalidOptionsException("The " + HIVE_DROP_DELIMS_ARG
                + " option conflicts with the " + HIVE_DELIMS_REPLACEMENT_ARG
                + " option." + HELP_STR);
        }

        // Make sure that one of hCatalog or hive jobs are used
        String hCatTable = options.getHCatTableName();
        if (hCatTable != null && options.doHiveImport()) {
            throw new InvalidOptionsException("The " + HCATALOG_TABLE_ARG
                + " option conflicts with the " + HIVE_IMPORT_ARG
                + " option." + HELP_STR);
        }

        if (options.doHiveImport()
            && options.getFileLayout() == SqoopOptions.FileLayout.AvroDataFile) {
            throw new InvalidOptionsException("Hive import is not compatible with "
                + "importing into AVRO format.");
        }

        if (options.doHiveImport()
            && options.getFileLayout() == SqoopOptions.FileLayout.SequenceFile) {
            throw new InvalidOptionsException("Hive import is not compatible with "
                + "importing into SequenceFile format.");
        }

        // Many users are reporting issues when they are trying to import data
        // directly into hive warehouse. This should prevent users from doing
        // so in case of a default location.
        String defaultHiveWarehouse = "/user/hive/warehouse";
        if (options.doHiveImport()
            && ((
            options.getWarehouseDir() != null
                && options.getWarehouseDir().startsWith(defaultHiveWarehouse)
        ) || (
            options.getTargetDir() != null
                && options.getTargetDir().startsWith(defaultHiveWarehouse)
        ))) {
            LOG.warn("It seems that you're doing hive import directly into default");
            LOG.warn("hive warehouse directory which is not supported. Sqoop is");
            LOG.warn("firstly importing data into separate directory and then");
            LOG.warn("inserting data into hive. Please consider removing");
            LOG.warn("--target-dir or --warehouse-dir into /user/hive/warehouse in");
            LOG.warn("case that you will detect any issues.");
        }

        // Warn about using hive specific arguments without hive import itself
        // In HCatalog support some of the Hive options are reused
        if (!options.doHiveImport()
            && (((options.getHiveHome() != null
            && !options.getHiveHome().equals(SqoopOptions.getHiveHomeDefault())
            && hCatTable == null))
            || options.doOverwriteHiveTable()
            || options.doFailIfHiveTableExists()
            || (options.getHiveTableName() != null
            && !options.getHiveTableName().equals(options.getTableName()))
            || (options.getHivePartitionKey() != null && hCatTable == null)
            || (options.getHivePartitionValue() != null && hCatTable == null)
            || (options.getMapColumnHive().size() > 0 && hCatTable == null))) {
            LOG.warn("It seems that you've specified at least one of following:");
            LOG.warn("\t--hive-home");
            LOG.warn("\t--hive-overwrite");
            LOG.warn("\t--create-hive-table");
            LOG.warn("\t--hive-table");
            LOG.warn("\t--hive-partition-key");
            LOG.warn("\t--hive-partition-value");
            LOG.warn("\t--map-column-hive");
            LOG.warn("Without specifying parameter --hive-import. Please note that");
            LOG.warn("those arguments will not be used in this session. Either");
            LOG.warn("specify --hive-import to apply them correctly or remove them");
            LOG.warn("from command line to remove this warning.");
            LOG.info("Please note that --hive-home, --hive-partition-key, ");
            LOG.info("\t hive-partition-value and --map-column-hive options are ");
            LOG.info("\t are also valid for HCatalog imports and exports");
        }
    }

    protected void validateAccumuloOptions(SqoopOptions options)
        throws InvalidOptionsException {
        if ((options.getAccumuloColFamily() != null
            && options.getAccumuloTable() == null)
            || (options.getAccumuloColFamily() == null
            && options.getAccumuloTable() != null)) {
            throw new InvalidOptionsException(
                "Both --accumulo-table and --accumulo-column-family must be set."
                    + HELP_STR);
        }

        if (options.getAccumuloTable() != null
            && options.getHBaseTable() != null) {
            throw new InvalidOptionsException("HBase import is incompatible with "
                + "Accumulo import.");
        }
        if (options.getAccumuloTable() != null
            && options.getFileLayout() != SqoopOptions.FileLayout.TextFile) {
            throw new InvalidOptionsException("Accumulo import is not compatible "
                + "with importing into file format.");
        }
        if (options.getAccumuloTable() != null
            && options.getHBaseColFamily() != null) {
            throw new InvalidOptionsException("Use --accumulo-column-family with "
                + "Accumulo import.");
        }
        if (options.getAccumuloTable() != null
            && options.getAccumuloUser() == null) {
            throw
                new InvalidOptionsException("Must specify Accumulo user.");
        }
        if (options.getAccumuloTable() != null
            && options.getAccumuloInstance() == null) {
            throw new
                InvalidOptionsException("Must specify Accumulo instance.");
        }
        if (options.getAccumuloTable() != null
            && options.getAccumuloZookeepers() == null) {
            throw new
                InvalidOptionsException("Must specify Zookeeper server(s).");
        }
    }

    protected void validateHCatalogOptions(SqoopOptions options)
        throws InvalidOptionsException {
        // Make sure that one of hCatalog or hive jobs are used
        String hCatTable = options.getHCatTableName();
        if (hCatTable == null) {
            if (options.getHCatHome() != null && !options.getHCatHome().equals(SqoopOptions.getHCatHomeDefault())) {
                LOG.warn("--hcatalog-home option will be ignored in "
                    + "non-HCatalog jobs");
            }
            if (options.getHCatDatabaseName() != null) {
                LOG.warn("--hcatalog-database option will be ignored  "
                    + "without --hcatalog-table");
            }

            if (options.getHCatStorageStanza() != null) {
                LOG.warn("--hcatalog-storage-stanza option will be ignored "
                    + "without --hatalog-table");
            }
            return;
        }

        if (options.explicitInputDelims()) {
            LOG.warn("Input field/record delimiter options are not "
                + "used in HCatalog jobs unless the format is text.   It is better "
                + "to use --hive-import in those cases.  For text formats");
        }
        if (options.explicitOutputDelims()
            || options.getHiveDelimsReplacement() != null
            || options.doHiveDropDelims()) {
            LOG.warn("Output field/record delimiter options are not useful"
                + " in HCatalog jobs for most of the output types except text based "
                + " formats is text. It is better "
                + "to use --hive-import in those cases.  For non text formats, ");
        }
        if (options.doHiveImport()) {
            throw new InvalidOptionsException("The " + HCATALOG_TABLE_ARG
                + " option conflicts with the " + HIVE_IMPORT_ARG
                + " option." + HELP_STR);
        }
        if (options.getTargetDir() != null) {
            throw new InvalidOptionsException("The " + TARGET_DIR_ARG
                + " option conflicts with the " + HCATALOG_TABLE_ARG
                + " option." + HELP_STR);
        }
        if (options.getWarehouseDir() != null) {
            throw new InvalidOptionsException("The " + WAREHOUSE_DIR_ARG
                + " option conflicts with the " + HCATALOG_TABLE_ARG
                + " option." + HELP_STR);
        }

        if (options.isAppendMode()) {
            throw new InvalidOptionsException("Append mode for imports is not "
                + " compatible with HCatalog. Please remove the parameter"
                + "--append-mode");
        }
        if (options.getExportDir() != null) {
            throw new InvalidOptionsException("The " + EXPORT_PATH_ARG
                + " option conflicts with the " + HCATALOG_TABLE_ARG
                + " option." + HELP_STR);
        }

        if (options.getFileLayout() == SqoopOptions.FileLayout.AvroDataFile) {
            throw new InvalidOptionsException("HCatalog job is not compatible with "
                + " AVRO format option " + FMT_AVRODATAFILE_ARG
                + " option." + HELP_STR);

        }

        if (options.getFileLayout() == SqoopOptions.FileLayout.SequenceFile) {
            throw new InvalidOptionsException("HCatalog job  is not compatible with "
                + "SequenceFile format option " + FMT_SEQUENCEFILE_ARG
                + " option." + HELP_STR);
        }

        if (options.getHCatalogPartitionKeys() != null
            && options.getHCatalogPartitionValues() == null) {
            throw new InvalidOptionsException("Either both --hcatalog-partition-keys"
                + " and --hcatalog-partition-values should be provided or both of these"
                + " options should be omitted.");
        }

        if (options.getHCatalogPartitionKeys() != null) {
            if (options.getHivePartitionKey() != null) {
                LOG.warn("Both --hcatalog-partition-keys and --hive-partition-key"
                    + "options are provided.  --hive-partition-key option will be"
                    + "ignored");
            }

            String[] keys = options.getHCatalogPartitionKeys().split(",");
            String[] vals = options.getHCatalogPartitionValues().split(",");

            if (keys.length != vals.length) {
                throw new InvalidOptionsException("Number of static partition keys "
                    + "provided dpes match the number of partition values");
            }

            for (int i = 0; i < keys.length; ++i) {
                String k = keys[i].trim();
                if (k.isEmpty()) {
                    throw new InvalidOptionsException(
                        "Invalid HCatalog static partition key at position " + i);
                }
            }
            for (int i = 0; i < vals.length; ++i) {
                String v = vals[i].trim();
                if (v.isEmpty()) {
                    throw new InvalidOptionsException(
                        "Invalid HCatalog static partition key at position " + v);
                }
            }
        } else {
            if (options.getHivePartitionKey() != null
                && options.getHivePartitionValue() == null) {
                throw new InvalidOptionsException("Either both --hive-partition-key and"
                    + " --hive-partition-value options should be provided or both of "
                    + "these options should be omitted");
            }
        }
        if (options.doCreateHCatalogTable() &&
            options.doDropAndCreateHCatalogTable()) {
            throw new InvalidOptionsException("Options --create-hcatalog-table" +
                " and --drop-and-create-hcatalog-table are mutually exclusive." +
                " Use any one of them");
        }
    }

    protected void validateHBaseOptions(SqoopOptions options)
        throws InvalidOptionsException {
        if ((options.getHBaseColFamily() != null && options.getHBaseTable() == null)
            || (options.getHBaseColFamily() == null
            && options.getHBaseTable() != null)) {
            throw new InvalidOptionsException(
                "Both --hbase-table and --column-family must be set together."
                    + HELP_STR);
        }

        if (options.isBulkLoadEnabled() && options.getHBaseTable() == null) {
            String validationMessage = String.format("Can't run import with %s "
                    + "without %s",
                BaseSqoopTool.HBASE_BULK_LOAD_ENABLED_ARG,
                BaseSqoopTool.HBASE_TABLE_ARG);
            throw new InvalidOptionsException(validationMessage);
        }
    }

    /**
     * Given an array of extra arguments (usually populated via this.extraArguments), determine the offset of the first
     * '--' argument in the list. Return 'extra.length' if there is none.
     */
    protected int getDashPosition(String[] extra) {
        int dashPos = extra.length;
        for (int i = 0; i < extra.length; i++) {
            if (extra[i].equals("--")) {
                dashPos = i;
                break;
            }
        }

        return dashPos;
    }
}
